package com.atguigu.edu.realtime.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.util.DimUtil;
import com.atguigu.edu.realtime.util.DruidPhoenixDSUtil;
import com.atguigu.edu.realtime.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.SQLException;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author Lec
 * @date 2022/8/30 14:32
 */

public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimJoinFunction<T> {
    private DruidDataSource dataSource = null;
    private ThreadPoolExecutor threadPoolExecutor = null;
    private String tableName = null;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        dataSource = DruidPhoenixDSUtil.getDataSource();
        threadPoolExecutor = ThreadPoolUtil.getThreadPoolExecutor();
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                // 获取连接
                DruidPooledConnection connection = null;
                try {
                    connection = dataSource.getConnection();
                } catch (SQLException e) {
                    e.printStackTrace();
                }

                // 拼接sql
                // select * from t where id = key
                // 读取对应维度表数据
//                String sql = "select * from "+ tableName + " where id = ";
                String id = getKey(input);
                JSONObject dimInfo = DimUtil.getDimInfo(connection, tableName, id);

                // 合并数据到input
                /*先判断diminfo不为空然后再去join*/
                if (dimInfo != null) {
                    join(input, dimInfo);
                }

                // 输出input
                resultFuture.complete(Collections.singleton(input));
            }
        });

    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        resultFuture.complete(Collections.singleton(input));
    }
}
